/*
 * Copyright (c) 2024 Huawei Technologies Co.,Ltd.
 *
 * openGauss is licensed under Mulan PSL v2.
 * You can use this software according to the terms and conditions of the Mulan PSL v2.
 * You may obtain a copy of Mulan PSL v2 at:
 *
 * http://license.coscl.org.cn/MulanPSL2
 *
 * THIS SOFTWARE IS PROVIDED ON AN "AS IS" BASIS, WITHOUT WARRANTIES OF ANY KIND,
 * EITHER EXPRESS OR IMPLIED, INCLUDING BUT NOT LIMITED TO NON-INFRINGEMENT,
 * MERCHANTABILITY OR FIT FOR A PARTICULAR PURPOSE.
 * See the Mulan PSL v2 for more details.
 * -------------------------------------------------------------------------
 *
 * htap_delta_table.cpp
 *
 *
 * IDENTIFICATION
 * 	  src/gausskernel/storage/htap/htap_delta_table.cpp
 *
 * -------------------------------------------------------------------------
 */

#include "row.h"
#include "txn.h"
#include "htap_logger.h"
#include "htap_helpers.h"
#include "htap_infra.h"
#include "htap_internal.h"
#include "htap_delta_table.h"
#include "utils/numeric.h"
#include "funcapi.h"
#include "global.h"
#include "column.h"
#include "table.h"
#include "htap_delete_map.h"
#include "access/imcs/imcs.h"
#include "access/imcs/imcu_cache_mgr.h"
#include "access/imcs/imcs_ctlg.h"
#include <new>
#include <string.h>

namespace htap {
HTAP_DECLARE_LOGGER(HTAPDelta, HTAP)


HTAPDelta::HTAPDelta(bool isFlush /* = false */, bool useTxn /* =true */)
    : m_minCtid(ULLONG_MAX),
      m_maxCtid(0),
      m_tableID(-1),
      m_csn(-1),
      m_lastFlushedCsn(-1),
      m_currTable(nullptr),
      m_hasData(false),
      m_isFlush(isFlush),
      m_useTxn(useTxn),
      m_newRecords(0),
      m_txn(nullptr),
      m_delIterIdx(0),
      m_scanIterIdx(0)
{}

HTAPDelta::~HTAPDelta() {}

void HTAPDelta::Destroy() {}

void HTAPDelta::reset()
{
    m_minCtid = ULLONG_MAX;
    m_maxCtid = 0;
    m_delIterIdx = 0;
    m_scanIterIdx = 0;
    m_currTable = nullptr;
    m_hasData = false;
    m_lastFlushedCsn = -1;
    m_csn = -1;
    m_newRecords = 0;
}

MOT::Row *HTAPDelta::getRow(MOT::Sentinel *sentinel) const
{
    MOT::Row *row = nullptr;
    if (m_txn != nullptr) {
        row = m_txn->m_accessMgr->ReadTxnRow(sentinel);
        if (row != nullptr) {
            return row;
        } /* Found the row in current transaction */
    }

    row = sentinel->GetVisibleRowVersion(m_csn);
    if (row != nullptr && (row->GetCommitSequenceNumber() >= m_lastFlushedCsn)) {
        return row;
    }

    return nullptr;
}

MOT::Row *HTAPDelta::getFirstRelevantRow(MOT::IndexIterator *cursor) const
{
    while (cursor->IsValid()) {
        MOT::Row *row = getRow(cursor->GetPrimarySentinel());
        if (row != nullptr) {
            return row;
        }
        cursor->Next();
    }
    return nullptr;
}


uint32_t HTAPDelta::addRecords()
{
    MOT::Index *ix = m_currTable->GetPrimaryIndex();
    uint32_t newRecords = 0;

    MOT::IndexIterator *cursor = ix->Begin((m_txn != nullptr ? m_txn->GetThdId() : 0));
    if (cursor == nullptr) {
        HTAP_LOG_ERROR("HTAPDelta failed to get cursor");
        return newRecords;
    }

    while (cursor->IsValid()) {
        MOT::Row *row = getRow(cursor->GetPrimarySentinel());
        if (row != nullptr) {
            newRecords++;
        }
        cursor->Next();
    }

    cursor->Destroy();
    delete cursor;
    return newRecords;
}

bool HTAPDelta::BuildIndex()
{
    if (m_useTxn) {
        if (!HTAPInfra::HTAPTxnEnsure()) {
            return false;
        }
    } else {
        if (!MOT::InitMasstreeThreadinfo()) {
            return false;
        }
    }

    if (unlikely(m_useTxn)) {
        m_txn = HTAPGetSafeTxn();
        if (m_txn == nullptr) {
            HTAP_LOG_ERROR("Failed to get TxnManager");
            return false;
        }
        m_txn->SetSnapshot();
        m_currTable = m_txn->GetTableByExternalId(m_tableID);
    } else {
        m_currTable = MOT::GetTableManager()->GetTableByExternal(m_tableID);
    }

    if (m_currTable == nullptr) {
        HTAP_LOG_ERROR("Failed to get Delta table for ID: %d", m_tableID);
        return false;
    }


    IMCSDesc *imcsDesc = IMCU_CACHE->GetImcsDesc(m_tableID);
    m_lastFlushedCsn = 0;
    if (imcsDesc != nullptr) {
        m_lastFlushedCsn = imcsDesc->last_flushed_csn;
        if (unlikely(m_txn != nullptr && m_lastFlushedCsn > 0 &&
            m_txn->GetGcSession()->GetCurrentEpoch() > m_lastFlushedCsn)) {
            m_txn->GetGcSession()->GcSetEpoch(m_lastFlushedCsn);
        }
    }

    MOT::Index *ix = m_currTable->GetPrimaryIndex();
    if (ix == nullptr) {
        HTAP_LOG_ERROR("Failed to get Delta primary index");
        return false;
    }
    if (setMin(ix) && setMax(ix)) {
        m_hasData = true;
    }

    return true;
}

bool HTAPDelta::BuildIndexOnDeltaTable(Oid tableID, CommitSeqNo csn)
{
    reset();
    m_tableID = tableID;
    m_csn = csn;

    if (IS_PGXC_DATANODE) {
        if (!BuildIndex()) {
            HTAP_LOG_ERROR("Failed initialize delta table");
            return false;
        }
    }

    HTAP_LOG_DEBUG("HTAPDelta create succeeded. OID: (%d), CSN: (%lld)", m_tableID, m_csn);
    return true;
}

bool HTAPDelta::buildWithAllCSNVersion(Oid tableID, CommitSeqNo csn)
{
    HTAP_LOG_DEBUG("HTAPDelta create succeeded. OID: (%d), CSN: (%lld)", tableID, csn);
    reset();
    m_tableID = tableID;
    m_csn = csn;

    if (IS_PGXC_DATANODE) {
        if (!BuildIndex()) {
            HTAP_LOG_ERROR("Failed initialize");
            return false;
        }
    }

    HTAP_LOG_DEBUG("HTAPDelta create succeeded. OID: (%d), CSN: (%lld)", m_tableID,
        m_csn);
    return true;
}

bool HTAPDelta::BuildIndexOnNewDeltaTable(Oid tableID)
{
    HTAP_LOG_DEBUG("HTAPDelta rebuild started. OID: (%d), CSN: (%lld)", tableID,
        m_csn);

    reset();
    m_tableID = tableID;

    if (IS_PGXC_DATANODE) {
        if (!BuildIndex()) {
            HTAP_LOG_ERROR("Failed initialize");
            return false;
        }
    }

    HTAP_LOG_DEBUG("HTAPDelta rebuild succeeded. OID: (%d), CSN: (%lld)", m_tableID,
        m_csn);

    return true;
}

bool HTAPDelta::setMin(MOT::Index *ix)
{
    MOT::IndexIterator *cursor = ix->Begin((m_txn != nullptr ? m_txn->GetThdId() : 0));
    if (cursor == nullptr) {
        HTAP_LOG_ERROR("Failed to get cursor");
        return false;
    }

    MOT::Row *row = getFirstRelevantRow(cursor);
    bool ret = false;

    if (row != nullptr) {
        ret = true;
        m_minCtid = row->GetCTIDFlat();
    }

    cursor->Destroy();
    delete cursor;
    return ret;
}

bool HTAPDelta::setMax(MOT::Index *ix)
{
    MOT::IndexIterator *cursor = ix->BeginReverse((m_txn != nullptr ? m_txn->GetThdId() : 0));
    if (cursor == nullptr) {
        HTAP_LOG_ERROR("Failed to get cursor");
        return false;
    }

    MOT::Row *row = getFirstRelevantRow(cursor);
    bool ret = false;

    if (row != nullptr) {
        ret = true;
        m_maxCtid = row->GetCTIDFlat();
    }

    cursor->Destroy();
    delete cursor;
    return ret;
}

bool HTAPDelta::checkIMCU(IMCUDesc *imcuDesc) const
{
    CTID *cu_min_val = (CTID *)&(imcuDesc->cu_min);
    CTID *cu_max_val = (CTID *)&(imcuDesc->cu_max);
    HTAPCTID minRowCtid(cu_min_val->item.ip_blkid.bi_hi, cu_min_val->item.ip_blkid.bi_lo, cu_min_val->item.ip_posid,
        HTAP_DELTA_FLUSH_TAG);
    HTAPCTID maxRowCtid(cu_max_val->item.ip_blkid.bi_hi, cu_max_val->item.ip_blkid.bi_lo, cu_max_val->item.ip_posid,
        HTAP_DELTA_FLUSH_TAG);

    return !(minRowCtid.GetCTIDFlat() > m_maxCtid || maxRowCtid.GetCTIDFlat() < m_minCtid);
}

bool HTAPDelta::checkDeltaStatus()
{
    if (!m_useTxn) {
        return false;
    }

    if (m_currTable == nullptr) {
        return true;
    }

    return m_txn->GetOidNodeInf(m_currTable->GetRelID(), false) != nullptr;
}

void HTAPDelta::initLookupCtx(LookupContext &ctx) const
{
    ctx.m_index = m_currTable->GetPrimaryIndex();
    ctx.m_key = (MOT::MaxKey *)ctx._buf;
    ctx.m_key->InitKey(ctx.m_index->GetKeyLength());
}

bool HTAPDelta::checkRowgroup(LookupContext &ctx, IMCUDesc *imcuDesc) const
{
    Assert(imcuDesc != nullptr);

    /* Get Flat Min/Max values */
    CTID *cu_min_val = (CTID *)&(imcuDesc->cu_min);
    CTID *cu_max_val = (CTID *)&(imcuDesc->cu_max);
    HTAPCTID minRowCtid(cu_min_val->item.ip_blkid.bi_hi, cu_min_val->item.ip_blkid.bi_lo, cu_min_val->item.ip_posid,
        HTAP_DELTA_FLUSH_TAG);
    HTAPCTID maxRowCtid(cu_max_val->item.ip_blkid.bi_hi, cu_max_val->item.ip_blkid.bi_lo, cu_max_val->item.ip_posid,
        HTAP_DELTA_FLUSH_TAG);
    uint64_t minCTIDKey = minRowCtid.GetCTIDKey();
    uint64_t maxFlatCTIDKey = maxRowCtid.GetCTIDFlat();

    ctx.m_key->CpKey((uint8_t *)(&minCTIDKey), ctx.m_index->GetKeyLength());
    bool found = false;

    /* Search for the exect CTID or bigger */
    MOT::IndexIterator *cursor =
        ctx.m_index->Search(ctx.m_key, true, true, (m_txn != nullptr ? m_txn->GetThdId() : 0), found);

    if (cursor == nullptr) {
        return false;
    }
    bool isRelevant = false;

    while (cursor->IsValid()) {
        const MOT::Key *foundKey = reinterpret_cast<const MOT::Key *>(cursor->GetKey());
        uint64_t *reversedKeyPtr = (uint64_t *)(foundKey->GetKeyBuf());
        uint64_t reversedKey = *reversedKeyPtr;
        uint64_t key = htobe64(reversedKey);

        if (key > maxFlatCTIDKey) {
            break;
        }
        MOT::Row *row = getRow(cursor->GetPrimarySentinel());
        if (row != nullptr) {
            isRelevant = true;
            break;
        }
        cursor->Next();
    }
    cursor->Destroy();
    delete cursor;
    return isRelevant;
}

bool HTAPDelta::lookUp(char &foundRecordType, LookupContext &ctx, const ItemPointerData &data) const
{
    HTAPCTID rowCtid(data.ip_blkid.bi_hi, data.ip_blkid.bi_lo, data.ip_posid, HTAP_DELTA_FLUSH_TAG);

    /* check ctid */
    if (!isCTIDInRange(rowCtid.GetCTIDFlat())) {
        return false;
    }

    uint64_t CTIDKey = rowCtid.GetCTIDKey();

    ctx.m_key->CpKey((uint8_t *)(&CTIDKey), ctx.m_index->GetKeyLength());

    MOT::Sentinel *sentinel = ctx.m_index->IndexReadSentinel(ctx.m_key, (m_txn != nullptr ? m_txn->GetThdId() : 0));
    if (sentinel == nullptr) {
        return false;
    }

    MOT::Row *row = getRow(sentinel);

    if (row == nullptr) {
        return false;
    }
    foundRecordType = row->GetCTIDStatus();

    return true;
}

// BuildBatchDelMask
bool HTAPDelta::BuildBatchDelMask(HTAPDeleteMap *batchDelMask, IMCU *ctid_imcu, IMCUDesc *imcuDesc,
    int ctid_imcu_rows, HTAPDeleteMap *delMask) const
{
    uint32_t removedByIMCSMap = 0;
    uint32_t removedByDelta = 0;

    if (ctid_imcu_rows > 0) {
        if (delMask != nullptr) {
            batchDelMask->copy(*delMask);
            removedByIMCSMap = batchDelMask->bitsSigned();
        } else {
            batchDelMask->reset();
        }

        if (m_hasData) {
            if (checkIMCU(imcuDesc)) {
                char foundRecordType;
                LookupContext ctx;
                initLookupCtx(ctx);

                if (checkRowgroup(ctx, imcuDesc)) {
                    for (uint32_t idx = 0; idx < (uint32_t)ctid_imcu_rows; idx++) {
                        if (delMask == nullptr || !batchDelMask->check(idx)) {
                            ScalarValue ctid_value = ctid_imcu->GetValue<sizeof(CTID), false>(idx);
                            CTID *ctid = (CTID *)(&ctid_value);

                            if (lookUp(foundRecordType, ctx, ctid->item)) {
                                removedByDelta++;
                                batchDelMask->set(idx);
                            }
                        }
                    }
                }
            }
        }
    } else {
        batchDelMask->reset();
    }

    HTAP_LOG_DEBUG("BuildBatchDelMask (Table: %d, Itr: %d, CUID: %d) - Input rows "
                   "number: (%d). Total Rows signed: (%d), by IMCS: (%d), by Delta: (%d)",
        m_tableID, m_delIterIdx, imcuDesc->cu_id, ctid_imcu_rows,
        removedByIMCSMap + removedByDelta, removedByIMCSMap, removedByDelta);
    m_delIterIdx++;
    return ((removedByIMCSMap + removedByDelta) > 0);
}

bool HTAPDelta::FillColumn(ScalarVector *colVec, int2 attlen, MOT::Column *col, Oid type, int deltaColIdx,
    uint32_t recordIdx, void *record) const
{
    bool isNull = false;
    MOT::Row *row = static_cast<MOT::Row *>(record);

    htap::HTAPAdaptor::MOTToDatum1(col, attlen, type, deltaColIdx, const_cast<uint8_t *>(row->GetData()), &isNull, colVec, recordIdx);

    colVec->ClearFlag(recordIdx);
    if (isNull) {
        colVec->SetNull(recordIdx);
    }

    colVec->m_rows++;

    return true;
}


bool HTAPDelta::FillRow(Datum *values, bool *nulls, size_t *lengths, const TupleDesc &imcs_rd_att,
    void *record) const
{
    uint32_t deltaColumnIdx = 0;
    for (int32_t currColumnIndx = 0; currColumnIndx < imcs_rd_att->natts; ++currColumnIndx) {
        if (imcs_rd_att->attrs[currColumnIndx].attisdropped) {
            HTAP_LOG_ERROR("HTAPDelta::FillBatch failed, cannot add dropped column. Column: "
                           "(%s), Table: (%s)",
                NameStr(imcs_rd_att->attrs[currColumnIndx].attname), m_currTable->GetTableName().c_str());
            continue;
        }

        MOT::Row *row = static_cast<MOT::Row *>(record);
        if (imcs_rd_att->attrs[currColumnIndx].attnum == VirtualCtidColID) {
            CTID reverted_ctid;
            errno_t rc = memset_s(&reverted_ctid, sizeof(CTID), 0, sizeof(CTID));
            securec_check(rc, "", "");
            HTAPAdaptor::FillCtid(row->GetCTIDFlat(), &reverted_ctid.item);
            HTAP_LOG_DEBUG("HTAPFlushWorker: Reverting %lld into %lld", row->GetCTIDFlat(), reverted_ctid);
            values[currColumnIndx] = *((ScalarValue *)&reverted_ctid);
            lengths[currColumnIndx] = SizeOfIptrData;
        } else {
            htap::HTAPAdaptor::MOTToDatum1(m_currTable->GetOrigTable()->GetField(deltaColumnIdx + 1),
                imcs_rd_att->attrs[currColumnIndx].attlen, imcs_rd_att->attrs[currColumnIndx].atttypid, deltaColumnIdx,
                const_cast<uint8_t *>(row->GetData()), &nulls[currColumnIndx], &values[currColumnIndx],
                &lengths[currColumnIndx]);

            ++deltaColumnIdx;
        }
    }

    return true;
}

void HTAPDelta::scanForFlush(IMCSBulkloadRows *batch_rows_ptr, Datum *values, bool *nulls, size_t *lengths,
    int2 *imcs_to_values, const TupleDesc &imcs_rd_att, ScanContext &ctx, uint32_t maxOutSize) const
{
    uint32_t recordIdx = 0;
    if (ctx.m_isFinished || m_currTable == nullptr || !m_hasData) {
        return;
    }

    if (ctx.m_cursor == nullptr) {
        if (!initScanContext(ctx)) {
            HTAP_LOG_ERROR("HTAPDelta::scanForFlush (isFlush: %d) - Failed to initialize context", m_isFlush);
            return;
        }
    }

    MOT::TxnManager *txn = HTAPGetSafeTxn();
    uint64_t epoch = MOT::GetCSNManager().GetGcEpoch();
    while (recordIdx < maxOutSize && ctx.m_cursor->IsValid()) {
        MOT::Sentinel *sentinel = ctx.m_cursor->GetPrimarySentinel();
        MOT::Row *row = getRow(sentinel);

        if (row != nullptr) {
            char type = row->GetCTIDStatus();
            if (type == HTAP_DELTA_INSERT_TAG || type == HTAP_DELTA_UPDATE_TAG) {
                FillRow(values, nulls, lengths, imcs_rd_att, row);
                batch_rows_ptr->append_one_tuple(
                    values,
                    nulls, lengths,
                    imcs_rd_att->natts - 1,
                    imcs_rd_att,
                    imcs_to_values,
                    imcs_rd_att->natts - 1);
                ++recordIdx;
            }

            (void)row->GetPrimarySentinel()->GetGcInfo().RefCountUpdate(MOT::AccessType::INC);
            txn->GcSessionRecordRcu(MOT::GC_QUEUE_TYPE::HTAP_DELETE_QUEUE,
                row->GetTable()->GetPrimaryIndex()->GetIndexId(), row, row->GetPrimarySentinel(),
                MOT::Row::DeleteRowDtorHtap, ROW_SIZE_FROM_POOL(row->GetTable()), epoch);
        }
        ctx.m_cursor->Next();
    }

    if (!ctx.m_cursor->IsValid()) {
        ctx.setFinished();
    }

    batch_rows_ptr->m_rows_curnum = (int64)recordIdx;

    HTAP_LOG_DEBUG(
        "HTAPDelta::scanForFlush (isFlush: %d) (Table: %d, Itr: %d) - Rows scanned (%d), Max requested (%d)",
        m_isFlush, m_tableID, m_scanIterIdx, recordIdx, maxOutSize);
    m_scanIterIdx++;
}

bool HTAPDelta::FillBatch(VectorBatch *vec_batch_out, const TupleDesc &rd_att, uint32_t &recordIdx,
    void *record, int32_t numberOfColumns, int32_t *colIDArray) const
{
    for (int32_t i = 1; i < numberOfColumns; ++i) {
        int currColumnIndx = colIDArray[i] - 1; /* idx in TupleDesc attrs */

        if (rd_att->attrs[currColumnIndx].attisdropped) {
            HTAP_LOG_ERROR("HTAPDelta::FillBatch (isFlush: %d) Failed - cannot add dropped column. Column: "
                           "(%s), Table: (%s)",
                m_isFlush, NameStr(rd_att->attrs[currColumnIndx].attname), m_currTable->GetTableName().c_str());
            return false;
        }
        if (currColumnIndx >= 0) { // we need to aply additional offset only for output vector position, all other
                                   // objects dont include CTID
            Assert(currColumnIndx < vec_batch_out->m_cols);
            ScalarVector *colVec = vec_batch_out->m_arr + currColumnIndx;
            int2 delta_col = m_currTable->GetOrigTable()->GetIMCSColumnIx(colIDArray[i]);
            if (!FillColumn(colVec, rd_att->attrs[currColumnIndx].attlen,
                m_currTable->GetOrigTable()->GetField(delta_col + 1), rd_att->attrs[currColumnIndx].atttypid, delta_col,
                recordIdx, record)) { // remove values from previews columns vec
                return false;
            }
        }
    }
    recordIdx++;
    return true;
}

bool HTAPDelta::initScanContext(ScanContext &ctx) const
{
    /* Set cursor to the relevant row */
    MOT::Index *ix = m_currTable->GetPrimaryIndex();
    if (ix == nullptr) {
        HTAP_LOG_ERROR("HTAPDelta::initScanContext (isFlush: %d) - Failed to get primery index", m_isFlush);
        return false;
    }
    if (m_isFlush) {
        MOT::MaxKey key;
        key.InitKey(ix->GetKeyLength());
        uint64_t ctidKey = ctx.m_startKey;
        key.CpKey((uint8_t *)(&ctidKey), ix->GetKeyLength());

        bool found = false;
        ctx.m_cursor = ix->Search(&key, true, true, 0, found, false);
        Assert(found);
    } else {
        ctx.m_cursor = ix->Begin((m_txn != nullptr ? m_txn->GetThdId() : 0));
    }

    if (ctx.m_cursor == nullptr) {
        HTAP_LOG_ERROR("HTAPDelta::initScanContext (isFlush: %d) - Failed to get cursor", m_isFlush);
        ctx.setFinished();
        return false;
    }

    return true;
}

uint64_t HTAPDelta::getNextCtid(ScanContext &ctx, uint32_t num_rows_to_advance) const
{
    HTAP_LOG_DEBUG("HTAPDelta::getNextCtid starting table %d, lastKey %d, num rows to advance %d",
        m_currTable->GetTableExId(), htobe64(ctx.m_startKey), num_rows_to_advance);

    MOT::Row *row = NULL;
    if (ctx.m_startKey == 0) {
        ctx.m_cursor = m_currTable->GetPrimaryIndex()->Begin(0);
        if (!ctx.m_cursor) {
            HTAP_LOG_ERROR("Failed to get cursor");
            ctx.setFinished();
            return false;
        }
        row = getFirstRelevantRow(ctx.m_cursor);
        if (row != NULL) {
            ctx.m_startKey = row->GetCTIDKey();
            HTAP_LOG_DEBUG("HTAPDelta::initScanContextForFlush for table %d finished, lastKey %d",
                m_currTable->GetTableExId(), htobe64(ctx.m_startKey));
            return ctx.m_startKey;
        }
        return 0;
    }

    /* Set cursor to the relevant row */
    Assert(ctx.m_cursor->IsValid());

    uint32_t current = 0;
    while (current < num_rows_to_advance && ctx.m_cursor->IsValid()) {
        row = getRow(ctx.m_cursor->GetPrimarySentinel());
        if (row != nullptr &&
            (row->GetCTIDStatus() == HTAP_DELTA_INSERT_TAG || row->GetCTIDStatus() == HTAP_DELTA_UPDATE_TAG)) {
            current++;
        }
        ctx.m_cursor->Next();
    }
    if (!ctx.m_cursor->IsValid())
        return 0;

    row = getRow(ctx.m_cursor->GetPrimarySentinel());
    while (ctx.m_cursor->IsValid() && row == nullptr) {
        ctx.m_cursor->Next();
        row = getRow(ctx.m_cursor->GetPrimarySentinel());
    }
    if (row != nullptr) {
        ctx.m_startKey = row->GetCTIDKey();
        HTAP_LOG_DEBUG("HTAPDelta::initScanContextForFlush for table %d finished, lastKey %d",
            m_currTable->GetTableExId(), htobe64(ctx.m_startKey));
        return ctx.m_startKey;
    }
    return 0;
}

void HTAPDelta::scan(VectorBatch *vec_batch_out, const TupleDesc &rd_att, ScanContext &ctx, uint32_t maxOutSize,
    int32_t numberOfColumns, int32_t *colIDArray) const
{
    uint32_t recordIdx = 0;

    if (ctx.m_isFinished || m_currTable == nullptr || !m_hasData) {
        return;
    } // if delta is empty

    if (ctx.m_cursor == nullptr) {
        if (!initScanContext(ctx)) {
            HTAP_LOG_ERROR("Failed to initialize context");
            return;
        }
    }

    while (recordIdx < maxOutSize && ctx.m_cursor->IsValid()) {
        MOT::Sentinel *sentinel = ctx.m_cursor->GetPrimarySentinel();
        MOT::Row *row = getRow(sentinel);

        if (row != nullptr) {
            char type = row->GetCTIDStatus();
            if (type == HTAP_DELTA_INSERT_TAG || type == HTAP_DELTA_UPDATE_TAG) {
                FillBatch(vec_batch_out, rd_att, recordIdx, row, numberOfColumns, colIDArray);
            }
        }
        ctx.m_cursor->Next();
    }

    if (!ctx.m_cursor->IsValid()) {
        ctx.setFinished();
    }

    vec_batch_out->m_rows = (int64)recordIdx;

    HTAP_LOG_DEBUG("HTAPDelta::scan (Table: %d, Itr: %d) - Rows scaned (%d), Max requested (%d)",
        m_tableID, m_scanIterIdx, recordIdx, maxOutSize);
    m_scanIterIdx++;
}

HeapTuple HTAPDelta::delta_scan_getnexttuple(StatsScanContext &ctx)
{
    if (ctx.m_isFinished || m_currTable == nullptr || !m_hasData) {
        return nullptr;
    } // if delta is empty

    if (ctx.m_cursor == nullptr) {
        if (!initScanContext(ctx)) {
            HTAP_LOG_ERROR("Failed to initialize context");
            return nullptr;
        }
    }

    if (ctx.m_row == nullptr) {
        while (ctx.m_row == nullptr && ctx.m_cursor->IsValid()) {
            MOT::Sentinel *sentinel = ctx.m_cursor->GetPrimarySentinel();
            // get first row version
            ctx.m_row = sentinel->GetData();
            ctx.m_cursor->Next();
        }
    }

    if (ctx.m_row == nullptr) {
        return nullptr;
    }

    TupleDesc tupdesc = CreateTemplateTupleDesc(Nnum_delta_fields, false, TableAmHeap);
    TupleDescInitEntry(tupdesc, (AttrNumber)Anum_delta_csn, "csn", INT8OID, -1, 0);
    TupleDescInitEntry(tupdesc, (AttrNumber)Anum_delta_rowid, "rowid", INT8OID, -1, 0);
    TupleDescInitEntry(tupdesc, (AttrNumber)Anum_delta_status, "status", CHAROID, -1, 0);
    BlessTupleDesc(tupdesc);

    Datum *values = (Datum *)palloc0(Nnum_delta_fields * sizeof(Datum));
    bool *nulls = (bool *)palloc0(Nnum_delta_fields * sizeof(bool));

    values[Anum_delta_csn - 1] = UInt64GetDatum(ctx.m_row->GetCommitSequenceNumber());
    nulls[Anum_delta_csn - 1] = false;

    values[Anum_delta_rowid - 1] = ctx.m_row->GetCTIDFlat();
    nulls[Anum_delta_rowid - 1] = false;

    values[Anum_delta_status - 1] = ctx.m_row->GetCTIDStatus();
    nulls[Anum_delta_status - 1] = false;

    // Get next row version
    ctx.m_row = ctx.m_row->GetNextVersion();

    if (!ctx.m_cursor->IsValid() && ctx.m_row == nullptr) {
        ctx.setFinished();
    }

    HeapTuple tuple = heap_form_tuple(tupdesc, values, nulls);
    pfree(values);
    pfree(nulls);
    return tuple;
}

HTAPDelta::ScanContext::ScanContext(uint64_t startKey /* = 0 */)
    : m_cursor(nullptr), m_isFinished(false), m_startKey(startKey)
{}

HTAPDelta::ScanContext::~ScanContext()
{
    reset();
}

void HTAPDelta::ScanContext::reset()
{
    if (m_cursor != nullptr) {
        m_cursor->Destroy();
        delete m_cursor;
        m_cursor = nullptr;
    }
    m_isFinished = false;
}

void HTAPDelta::ScanContext::setFinished()
{
    if (m_cursor != nullptr) {
        m_cursor->Destroy();
        delete m_cursor;
        m_cursor = nullptr;
    }
    m_isFinished = true;
}

HTAPDelta::StatsScanContext::StatsScanContext() : m_row(nullptr) {}

HTAPDelta::StatsScanContext::~StatsScanContext() {}
}
